package com.mxd.flink.connector.table;

import com.mxd.flink.connector.config.RedisCmd;
import com.mxd.flink.connector.config.RedisOptions;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.format.EncodingFormat;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.sink.SinkFunctionProvider;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.RowKind;

import java.util.List;

/**
 * @author rongdi
 * @date 2022/9/18 19:50
 */
public class RedisDynamicTableSink implements DynamicTableSink {

    private ReadableConfig config;

    /**
     * 用于根据format配置项指定的格式化方式和表结构序列化成对应格式的对象
     */
    private EncodingFormat<SerializationSchema<RowData>> encodingFormat;

    private ResolvedSchema resolvedSchema;

    /**
     * SinkFunction的并行度
     */
    private Integer sinkParallelism;

    /**
     * 配置项中的command
     */
    private String command;

    public RedisDynamicTableSink(ResolvedSchema resolvedSchema, ReadableConfig config, EncodingFormat<SerializationSchema<RowData>> encodingFormat) {
        this.resolvedSchema = resolvedSchema;
        this.config = config;
        this.encodingFormat = encodingFormat;
        this.sinkParallelism = config.get(RedisOptions.SINK_PARALLELISM);
        this.command = config.get(RedisOptions.COMMAND);
    }

    /**
     * getChangelogMode()方法需要返回该Sink可以接受的change log行的类别。由于向Redis写入的数据可以是只追加的，
     * 也可以是带有回撤语义的（如各种聚合数据），因此支持INSERT、DELETE和UPDATE_AFTER类别。
     * @param changelogMode
     * @return
     */
    @Override
    public ChangelogMode getChangelogMode(ChangelogMode changelogMode) {
        /**
         * 如果是redis的lpush、rpush、sadd命令，由于命令特性只支持insert语义，不支持更新
         */
        if(RedisCmd.LPUSH.equalsIgnoreCase(command) || RedisCmd.RPUSH.equalsIgnoreCase(command) || RedisCmd.SADD.equalsIgnoreCase(command)) {
            return ChangelogMode.newBuilder().addContainedKind(RowKind.INSERT).build();
        }
        return ChangelogMode.newBuilder().addContainedKind(RowKind.INSERT).addContainedKind(RowKind.UPDATE_AFTER).build();
    }

    @Override
    public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
        /**
         * 拿到数据类型对象，也就是表结构里每个字段的名称、类型等信息
         */
        DataType dataType = this.resolvedSchema.toPhysicalRowDataType();
        /**
         * 这里由于直接传tableSchema会因为没法序列化而报错，所以这里生成一个可序列化的schema传过去
         */
        SerializationSchema<RowData> serializationSchema = encodingFormat.createRuntimeEncoder(context, dataType);
        /**
         * 获取主键列，serializationSchema中找不到主键信息，只能在tableSchema对象中找了
         */
        List<String> primaryKeys = this.resolvedSchema.getPrimaryKey().isPresent() ? this.resolvedSchema.getPrimaryKey().get().getColumns() : null;
        return SinkFunctionProvider.of(new RedisSinkFunction(this.config, serializationSchema, dataType, primaryKeys),this.sinkParallelism);
    }

    @Override
    public DynamicTableSink copy() {
        return new RedisDynamicTableSink(this.resolvedSchema, this.config,this.encodingFormat);
    }

    @Override
    public String asSummaryString() {
        return "Redis Dynamic Table Sink";
    }

}
